package com.pw.study.flink.window;

import com.pw.study.flink.entities.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


public class AggWindow {
    public static void main(String[] args) {

    }
    private static void dealAgg() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        env.socketTextStream("hadoop111",9999).map(v -> {
            String[] line = v.split(" ");
            return new WaterSensor(line[0],Long.valueOf(line[1]),Integer.valueOf(line[2]));
        }).keyBy(WaterSensor::getId).window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .aggregate(new AggregateFunction<WaterSensor, Avg, Double>() {
                    @Override
                    public Avg createAccumulator() {
                        System.out.println("createAccumulator..............");
                        return new Avg();
                    }

                    @Override
                    public Avg add(WaterSensor sensor, Avg acc) {
                        System.out.println("add ................");
                        acc.vcSum += sensor.getVc();
                        acc.count++;
                        return acc;
                    }

                    @Override
                    public Double getResult(Avg acc) {
                        System.out.println("result ......");
                        return acc.vcSum * 1.0 / acc.count;
                    }

                    @Override
                    public Avg merge(Avg avg, Avg acc1) {
                        System.out.println("merge...........");
                        return null;
                    }
                }, new ProcessWindowFunction<Double, String, String, TimeWindow>() {
                    @Override
                    public void process(String key, Context ctx, Iterable<Double> elements, Collector<String> out) throws Exception {
                        Double result = elements.iterator().next();
                        out.collect(key + "的平均值: " + result + "  " + ctx.window());
                    }
                }).print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    public static class Avg {
        public Integer vcSum = 0;
        public Long count = 0L;

    }
}
